import java.util.concurrent.*

THREADS = 5
pool = Executors.newFixedThreadPool(THREADS)
defer = { c -> pool.submit(c as Callable) }
new File(directory).eachFileMatch(patternOfFilesToParse) {
    f ->
    defer{f.eachLine { ln ->   pipeline(ln) }}
}
def unit= java.util.concurrent.TimeUnit.SECONDS 
pool.shutdown()
pool.awaitTermination(3600, unit) 
